package cn.doitedu.dynamic_rule.demos.timer;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by 31464 on 2021/8/24.
 */
public class SensorKeyedProcessFunction extends KeyedProcessFunction<String,SensorReading,String> {

    ValueState<SensorReading> lastState;

    ValueState<Long>  currentTimer;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<SensorReading> lastTempDescriptor = new ValueStateDescriptor<>("lastTemp", SensorReading.class);
        ValueStateDescriptor<Long> currentTimerDescriptor = new ValueStateDescriptor<>("currentTimer", Long.class);
        // 定义一个状态，用来保存上一个数据的温度值
        lastState = getRuntimeContext().getState(lastTempDescriptor);
        // 定义一个状态，用来保存定时器的时间戳
        currentTimer = getRuntimeContext().getState(currentTimerDescriptor);
    }

    @Override
    public void processElement(SensorReading sensorReading, Context ctx, Collector<String> out) throws Exception {
        //上一条数据的温度值
        SensorReading lastSensorReading= lastState.value();
        //更新温度值
        lastState.update(sensorReading);

        //当前定时器的时间戳
        Long currentTimerTs = currentTimer.value();


        /**
         * preTemp==null 则表示为第一条数据，无需出发定时器或者预警
         * sensorReading.getTemperature()<=preTemp   表示温度在下降或维持 无需出发定时器或者预警
         */
        if(lastSensorReading==null || sensorReading.getTemperature()<=lastSensorReading.getTemperature()){
            if(currentTimerTs!=null) {
                ctx.timerService().deleteProcessingTimeTimer(currentTimerTs);
            }
        }else{
            // 温度上升且没有设过定时器，则注册定时器
            long timerTs=ctx.timerService().currentProcessingTime() + 5000L;
            ctx.timerService().registerProcessingTimeTimer( timerTs );
            currentTimer.update(timerTs);
        }


    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
        out.collect(ctx.getCurrentKey()+",温度在持续上升----");
        currentTimer.clear();
    }
}
